package com.bd08.flink.demo.renrenren

import org.apache.flink.api.scala.ExecutionEnvironment
import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.api.scala.createTypeInformation

object Customers extends App {
  case class Customer(id:String,name:String,address:String)
  val env = ExecutionEnvironment.getExecutionEnvironment

  val csvInput = env.readCsvFile[(Customer)]("C:\\Users\\任仕豪\\Desktop\\游戏\\buggeikan\\src\\main\\resources\\customers.csv", includedFields = Array(1,2,6),quoteCharacter = '"')
  csvInput
    .filter(customer => customer.address.contains("Chicago"))
    .writeAsCsv("C:\\Users\\任仕豪\\Desktop\\游戏\\buggeikan\\src\\main\\resources\\Chiago","\n",",",FileSystem.WriteMode.OVERWRITE)

  csvInput.groupBy("name","address")
    .reduceGroup(group => {
      val e= group.next()
      (e.name+"|"+e.address,group.size)
    }).writeAsCsv("C:\\Users\\任仕豪\\Desktop\\游戏\\buggeikan\\src\\main\\resources\\Chiago2","\n",",",FileSystem.WriteMode.OVERWRITE)


  env.execute()
}
